package com.hmc.kafka.consumer.group;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.RoundRobinAssignor;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.time.Duration;
import java.util.Arrays;
import java.util.List;
import java.util.Properties;

public class CustomConsumer1 {
    public static void main(String[] args) {
        //创建配置类
        Properties properties = new Properties();
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop102:9092,hadoop103:9092");
        //反序列化
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,StringDeserializer.class.getName());
        //配置组id
        properties.put(ConsumerConfig.GROUP_ID_CONFIG,"0");
        //配置分区分配策略
        properties.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG, RoundRobinAssignor.class.getName());
        //创建消费者
        KafkaConsumer<String,String> consumer = new KafkaConsumer(properties);
        //订阅topic
        List<String> list = Arrays.asList("second","third");
        consumer.subscribe(list);
        //消费数据
        while (true) {
            //拉取数据，每一秒钟一次
            ConsumerRecords<String, String> message = consumer.poll(Duration.ofSeconds(1));
            message.iterator().forEachRemaining(a->{
                System.out.println(a);
            });
        }
    }
}
